#include <pthread.h>
#include <time.h>  // clock_gettime
#include <errno.h> // ETIMEDOUT

#include <iostream>
#include <string>
#include "LinkMqttClient.h"

using namespace controller;

LinkMqttClient::LinkMqttClient() {}
LinkMqttClient::~LinkMqttClient() { this->release(); }
bool LinkMqttClient::release()
{
    if (this->fd != nullptr)
    {
        std::cout << this->name() << "释放,m_isThreadRunning:" << std::to_string(m_isThreadRunning) << " thread:" << std::to_string(thread) << std::endl;
        if (m_isThreadRunning && thread)
        {
            m_isThreadRunning = false;
            pthread_join(thread, NULL);
            thread = 0;
        }
        MQTTClient_disconnect(this->fd, 1000);
        MQTTClient_destroy(&this->fd);
        this->fd = nullptr;
    }
    return true;
}

static bool _sub(std::string name, MQTTClient fd, std::string sub_topic, int qos)
{
    int rc = MQTTClient_subscribe(fd, sub_topic.c_str(), qos);
    if (rc != MQTTCLIENT_SUCCESS && rc != qos)
    {
        std::cout << name << "订阅 Err,主题:" << sub_topic << "code:" << std::to_string(rc) << std::endl;
        return false;
    }
    else
    {
        std::cout << name << "订阅 OK,主题:" << sub_topic << std::endl;
        return true;
    }
    return false;
}

// 读写接口
bool LinkMqttClient::open(std::map<std::string, std::string> arg)
{
    int rc = 0;
    if (arg.find("url") == arg.end())
    {
        std::cout << "LinkMqttClient 未设置 url" << std::endl;
        return false;
    }
    if (arg.find("deviceid") == arg.end())
    {
        std::cout << "LinkMqttClient 未设置 deviceid" << std::endl;
        return false;
    }
    if (arg.find("key") == arg.end())
    {
        std::cout << "LinkMqttClient 未设置 key" << std::endl;
        return false;
    }
    url = arg["url"];
    deviceid = arg["deviceid"];
    key = arg["key"];

    // 1. 初始化 options
    MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer;
    std::cout << "url:" << url << " device-id:" << deviceid << " key:" << key << std::endl;
    opts.username = deviceid.c_str();
    opts.password = key.c_str();
    opts.keepAliveInterval = 2; // <=5 s
    opts.cleansession = 1;

    // 2. 创建mqtt对象
    if ((rc = MQTTClient_create(&this->fd, url.c_str(), deviceid.c_str(), MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS)
    {
        std::cout << this->name() << "LinkMqttClient,无法创建对象,code=" << std::to_string(rc) << std::endl;
        return false;
    }

    // 3.连接mqtt服务器
    if ((rc = MQTTClient_connect(this->fd, &opts)) != MQTTCLIENT_SUCCESS)
    {
        std::cout << this->name() << "LinkMqttClient,连接失败,code = " << std::to_string(rc) << std::endl;
        return false;
    }
    if (this->OnLinkConnect)
        this->OnLinkConnect(this);

    this->_param = arg;

    // 4.订阅消息
    for (const auto &pair : arg)
    {
        std::string n = pair.first;
        std::string v = pair.second;
        if (n.find("sub") != std::string::npos) // 使用pair.first访问键，pair.second访问值
        {
            _sub(this->name(), this->fd, v, 1); // 默认使用 1,如有需求再做扩展
        }
    }

    // 5. 创建线程，监控消息
    m_isThreadRunning = true;
    if (thread == 0)
    {
        rc = pthread_create(&thread, NULL, LinkMqttThreadMonitor, this);
        if (rc != 0)
        {
            thread = 0;
            std::cout << name() << "无法启动接收线程，退出" << std::endl;
            this->release();
        }
    }
    return true;
}

bool LinkMqttClient::isThreadRunning() { return m_isThreadRunning; }
MQTTClient LinkMqttClient::client() { return fd; }
std::map<std::string, std::string> LinkMqttClient::params() { return this->_param; }
void *LinkMqttClient::LinkMqttThreadMonitor(void *pParam)
{
    LinkMqttClient *p_base = (LinkMqttClient *)pParam;
    if (p_base)
    {
        for (; p_base->isThreadRunning();)
        {
            char *topicName = NULL;
            int topicLen;
            MQTTClient_message *message = NULL;
            int rc = MQTTClient_receive(p_base->client(), &topicName, &topicLen, &message, 1000);
            if (rc == MQTTCLIENT_DISCONNECTED)
            {
                std::cout << p_base->name() << "mqtt连接已经断开,5秒后将重新连接" << std::endl;
                if (p_base->OnLinkDisconnect)
                {
                    p_base->OnLinkDisconnect(p_base);
                }
                // p_base->release();
                sleep(5);
                p_base->open(p_base->params());
            }
            else if (message)
            {
                LinkBuffer buf(message->payload, message->payloadlen);
                std::cout << p_base->name() << "<<(" << std::to_string(buf.size()) << "):" << p_base->dump_buffer(&buf) << std::endl
                          << "topic=" << topicName << std::endl;
                if (p_base->OnLinkRecv)
                {
                    p_base->OnLinkRecv(p_base, &buf, topicName);
                }
                MQTTClient_freeMessage(&message);
                MQTTClient_free(topicName);
            }
        }
        std::cout << p_base->name() << "线程退出." << std::endl;
    }
    else
    {
        // point null
    }

    pthread_exit(NULL);
}

bool LinkMqttClient::close()
{
    this->release();
    return true;
}

int LinkMqttClient::write(char *data, int len, const char *arg)
{
    int qos = 1;
    const char *topic = arg;
    LinkBuffer buf(data, len);
    std::cout << this->name() << ">>(" << std::to_string(buf.size()) << "):" << this->dump_buffer(&buf) << std::endl
              << "topic=" << std::string(topic) << std::endl;
    if (this->OnLinkSend)
        this->OnLinkSend(this, &buf, topic);
    MQTTClient_waitForCompletion(fd, last_token, 3000);
    int r = MQTTClient_publish(fd, topic, len, (void *)data, qos, 0, &last_token);
    return r;
}
// int LinkMqttClient::read(char *dataout, int len)
// {
//     return 0;
// }

// 状态
bool LinkMqttClient::is_open()
{
    return MQTTClient_isConnected(fd);
}
std::string LinkMqttClient::name()
{
    return ("mqtt[" + this->url + "]");
}
std::string LinkMqttClient::info()
{
    return "mqtt-c";
}